package com.moses.cityTraffic.monitor

import java.util.Properties

import com.moses.cityTraffic.utils._
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.datastream.BroadcastStream
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

/**
 * 实时车辆超速监控
 */
object OutOfSpeedMonitorAnalusis {

  def main(args: Array[String]): Unit = {

    // 创建流计算初始化对象
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    // stream2是从Mysql数据库中读取的卡口限速信息，特点：数据量少，更新不频繁
    val stream2: BroadcastStream[MonitorInfo] =
      streamEnv.addSource(new JdbcReadDataSource[MonitorInfo](classOf[MonitorInfo])).broadcast(GlobalConstants.MONITOR_STATE_DESCRIPTOR)

    // 配置kafka
    val props = new Properties()
    props.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092")
    props.setProperty("group.id", "moses_001")

    // 创建一个kafka的source
    // stream1是海量的数据流，不可以存入广播状态流中
    val stream1: DataStream[TrafficInfo] = streamEnv.addSource(new FlinkKafkaConsumer[String]("t_traffic", new SimpleStringSchema(), props).setStartFromEarliest())
//      val stream1: DataStream[TrafficInfo] = streamEnv.socketTextStream("hadoop102", 9999)
      .map((line: String) => {
      val arr: Array[String] = line.split(",")
      TrafficInfo(arr(0).toLong, arr(1), arr(2), arr(3), arr(4).toDouble, arr(5), arr(6))
    })

    // Flink中有connect和join做联结
    // join连接时，必须是两个一样类型的流
    // join适合内连接，有条件的连接，connect适合内连接和外连接
    stream1.connect(stream2)
      .process(new BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfLimitSpeedInfo] {
        /**
         * 连接处理 车辆经过速度和限速信息
         * @param in1
         * @param readOnlyContext
         * @param collector
         */
        override def processElement(in1: TrafficInfo, readOnlyContext: BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfLimitSpeedInfo]#ReadOnlyContext, collector: Collector[OutOfLimitSpeedInfo]) = {
          // 先从状态中得到当前卡口的限速信息
          val info: MonitorInfo = readOnlyContext.getBroadcastState(GlobalConstants.MONITOR_STATE_DESCRIPTOR).get(in1.monitorId)
          if(info != null) { // 表示当前这辆车经过的卡口是有限速的
            val limitSpeed: Int = info.limitSpeed
            val realSpeed: Double = in1.speed
            if(limitSpeed * 1.1 < realSpeed) {// 当前车辆超速通过卡口
              collector.collect(OutOfLimitSpeedInfo(in1.car, in1.monitorId, in1.roadId, realSpeed, limitSpeed, in1.actionTime))
            }
          }
        }

        /**
         * 处理广播流
         * @param in2
         * @param context
         * @param collector
         */
        override def processBroadcastElement(in2: MonitorInfo, context: BroadcastProcessFunction[TrafficInfo, MonitorInfo, OutOfLimitSpeedInfo]#Context, collector: Collector[OutOfLimitSpeedInfo]) = {
          // 把广播流中的数据保存到状态中
          context.getBroadcastState(GlobalConstants.MONITOR_STATE_DESCRIPTOR).put(in2.monitorId, in2)
        }
      }).addSink(new WriteDataSink[OutOfLimitSpeedInfo](classOf[OutOfLimitSpeedInfo]))

    streamEnv.execute()

  }
}
